package gupao.concurrency.threadpool.futures;

import gupao.concurrency.utils.ThreadPoolService;
import gupao.concurrency.utils.Utils;
import lombok.extern.slf4j.Slf4j;

import java.text.DecimalFormat;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Supplier;

@Slf4j
public class CompletableFutureApiSample {

    static ThreadPoolExecutor executor = ThreadPoolService.getOrCreateThreadPoolExecutor("price", 4, 8);

    public static void main(String[] args) throws Exception {
        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "8");
//        useLikeFuture();
        thenTest();
//        handleTest();
//        exceptionallyTest();
//        runAfterBoth();
//        runAfterEither();
//        applyToEither();
    }

    static void useAsFuture() throws Exception {
        CompletableFuture<Void> runAsyncAndGet = CompletableFuture.runAsync(() -> {
            log.info("run async...");
            Utils.sleepIgnoreExp(2);
            log.info("run async...completed.");
        });
        Utils.sleepIgnoreExp(800);
        runAsyncAndGet.cancel(true);
        runAsyncAndGet.get();
//        runAsyncAndGet.get(1, TimeUnit.SECONDS); //timeout exp

        CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> {
            log.info("run supplyAsync...");
            Utils.sleepIgnoreExp(2);
            log.info("run supplyAsync...completed.");
            return "Success";
        });

        supplyAsync.cancel(true);// if cancel called,then get result will exp
        String result = supplyAsync.get();
//        String result = supplyAsync.get(1, TimeUnit.SECONDS); //timeout exp
        log.info("supplyAsync completed with result: {}", result);
    }

    static Supplier<Double> priceSupplier = () -> {
        log.info("supplyAsync: query price...");
        Utils.sleepIgnoreExp(200);
        double price = new Random().nextDouble() * 100;
        log.info("supplyAsync: query price = {}...completed.", price);
        return price;
    };

    static Supplier<Integer> quantitySupplier = () -> {
        log.info("supplyAsync: query quantity...");
        Utils.sleepIgnoreExp(200);
        int quantity = new Random().nextInt(100);
        log.info("supplyAsync: query quantity = {} ...completed.", quantity);
        return quantity;
    };

    static BiConsumer<Double, Integer> moneyCalcConsumer = (price, quantity) -> {
        log.info("calc money and send to MQ, price: {}, quantity: {}", price, quantity);
        double money = price * quantity;
        //send MQ ...
        log.info("send to MQ, price: {}, quantity: {} => {}", price, quantity, money);
    };

    static void thenTest() throws ExecutionException, InterruptedException {
        CompletableFuture<Double> priceQueryFuture = CompletableFuture.supplyAsync(priceSupplier);
//        CompletableFuture<Integer> quantityQueryFuture = CompletableFuture.supplyAsync(quantitySupplier);

        /*System.out.println("thenAccept:");
        priceQueryFuture.thenAccept(CompletableFutureApiSample::printAsLog).get();
        priceQueryFuture.thenAcceptAsync(CompletableFutureApiSample::printAsLog).get();
        priceQueryFuture.thenAcceptAsync(CompletableFutureApiSample::printAsLog, executor).get(); //切换线程池
        priceQueryFuture.thenAcceptBoth(quantityQueryFuture, moneyCalcConsumer).get();
        priceQueryFuture.thenAcceptBothAsync(quantityQueryFuture, moneyCalcConsumer).get();
        priceQueryFuture.thenAcceptBothAsync(quantityQueryFuture, moneyCalcConsumer, executor).get();*/

        System.out.println("thenApply:");
        priceQueryFuture.thenApply(CompletableFutureApiSample::truncation).join();
        priceQueryFuture.thenApplyAsync(CompletableFutureApiSample::truncation).join();
        priceQueryFuture.thenApplyAsync(CompletableFutureApiSample::truncation, executor).join();//切换线程池

        /*System.out.println("thenCombine:");
        BiFunction<Integer, Double, Double> moneyCalcFunc = (quantity, price) -> {
            double money = quantity * price;
            log.info("calc quantity * price, {} * {} = {}", quantity, price, money);
            return money;
        };
        quantityQueryFuture.thenCombine(priceQueryFuture, moneyCalcFunc).get();
        quantityQueryFuture.thenCombineAsync(priceQueryFuture, moneyCalcFunc).get();
        quantityQueryFuture.thenCombineAsync(priceQueryFuture, moneyCalcFunc, executor).get();*/

        /*System.out.println("thenCompose:");
        Function<Integer, CompletionStage<Double>> composeFunc = quantity -> priceQueryFuture.thenApply(price -> {
            double money = truncation(price * quantity);
            log.info("compose to calc money: price({}) * quantity({}) = {}", price, quantity, money);
            return money;
        });
        quantityQueryFuture.thenCompose(composeFunc).get();
        quantityQueryFuture.thenComposeAsync(composeFunc).get();
        quantityQueryFuture.thenComposeAsync(composeFunc, executor).get();*/

        /*System.out.println("thenRun:");
        Runnable thenRunner = () -> {
            log.info("I can do nothing for the last step result because nobody give me the result.");
        };
        quantityQueryFuture.thenRun(thenRunner).get();
        quantityQueryFuture.thenRunAsync(thenRunner).get();
        quantityQueryFuture.thenRunAsync(thenRunner, executor).get();*/


    }

    static CompletableFuture<Integer> one = CompletableFuture.supplyAsync(() -> {
        log.info("one starting..");
        Utils.sleepIgnoreExp(new Random().nextInt(5000));
        int x = new Random().nextInt(100);
        log.info("one will completed.And return {}", x);
        return x;
    });
    static CompletableFuture<Integer> other = CompletableFuture.supplyAsync(() -> {
        log.info("other starting..");
        Utils.sleepIgnoreExp(new Random().nextInt(5000));
        int x = new Random().nextInt(100);
        log.info("one will completed.And return {}", x);
        return x;
    });
    static void applyToEither() {
        int result = one.applyToEither(other, resultFromEither -> {
            int x = resultFromEither * 3 + 5 ;
            log.info("apply from either, {} -> {}", resultFromEither, x);
            return x;
        }).join();
        log.info("Get result: {}", result);
    }

    static void acceptToEither(){
        one.acceptEither(other, resultFromEither -> {
            int x = resultFromEither * 3 + 5 ;
            log.info("apply from either, {} -> {}", resultFromEither, x);
        }).join();
    }

    static void runAfterEither() {
        AtomicInteger result = new AtomicInteger(0);
        CompletableFuture<Void> one = CompletableFuture.runAsync(() -> {
            log.info("one starting..");
            Utils.sleepIgnoreExp(new Random().nextInt(5000));
            int x = new Random().nextInt(100);
            if (result.compareAndSet(0, x)) {
                log.info("result:{} is set by CF one", x);
            }
            log.info("one will completed.");
        });
        CompletableFuture<Void> other = CompletableFuture.runAsync(() -> {
            log.info("other starting..");
            Utils.sleepIgnoreExp(new Random().nextInt(5000));
            int x = new Random().nextInt(100);
            if (result.compareAndSet(0, x)) {
                log.info("result:{} is set by CF other", x);
            }
            log.info("other will completed.");
        });
        one.runAfterEither(other, () -> {
            log.info("Either action: result = {}", result.get());
        }).join();
    }

    static void runAfterBoth() {
        AtomicInteger quantity = new AtomicInteger(0);
        AtomicReference<Double> price = new AtomicReference<>();
        AtomicReference<Double> money = new AtomicReference<>();
        CompletableFuture<Void> quantityQueryFuture = CompletableFuture.runAsync(() -> {
            quantity.set(quantitySupplier.get());
        });
        CompletableFuture<Void> priceQueryFuture = CompletableFuture.runAsync(() -> {
            price.set(truncation(priceSupplier.get()));
        });
        quantityQueryFuture.runAfterBothAsync(priceQueryFuture, () -> {
            money.set(price.get() * quantity.get());
            log.info("price: {}, quantity: {} has retrieved. Calc Money: {}",
                    price.get(), quantity.get(), money.get());
        }, executor).join();
    /*quantityQueryFuture.runAfterBoth(priceQueryFuture, () -> {
            money.set(price.get() * quantity.get());
            log.info("price: {}, quantity: {} has retrieved. Calc Money: {}", price.get(), quantity.get(), money.get());
        }).join();
        quantityQueryFuture.runAfterBothAsync(priceQueryFuture, () -> {
            money.set(price.get() * quantity.get());
            log.info("price: {}, quantity: {} has retrieved. Calc Money: {}", price.get(), quantity.get(), money.get());
        }).join();*/
    }

    static void handleTest() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> quantityQueryFuture = CompletableFuture.supplyAsync(quantitySupplier);
        BiFunction<Double, Throwable, Double> expFallbackHandler = (price, t) -> {
            if (t == null) {
                log.info("price is no exp, {}", price);
                return price;
            }
            double fallbackResult = fallbackMoney();
            log.error("exp occur, return fallback result: {}.", fallbackResult, t);
            return fallbackResult;
        };

        CompletableFuture.supplyAsync(() -> {
                    double price = priceSupplier.get();
                    if (price < 90) {
                        throw new RuntimeException("Price too low");
                    }
                    return price;
                }).handle(expFallbackHandler)
                .thenCombine(quantityQueryFuture, (price, quantity) -> {
                    double money = price * quantity;
                    log.info("calc: price({}) * quantity({}) = {}", price, quantity, money);
                    return truncation(money);
                }).whenComplete((result, t) -> {
                    log.info("Completed: result: {}", t == null ? result :
                            (t.getClass().getName() + "." + t.getMessage()));
                }).get();
        /*log.info("result: {}", moneyCalcFuture.get());*/
    }

    static void exceptionallyTest() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> quantityQueryFuture = CompletableFuture
                .supplyAsync(quantitySupplier);
        CompletableFuture.supplyAsync(() -> {
                    double price = priceSupplier.get();
                    if (price < 90) {
                        throw new RuntimeException("Price too low");
                    }
                    return price;
                }).exceptionally(t -> {
                    log.error("exception {} occur, use fallback", t.getClass().getName());
                    return fallbackMoney();
                })
                .thenCombine(quantityQueryFuture, (price, quantity) -> {
                    double money = price * quantity;
                    log.info("calc: price({}) * quantity({}) = {}", price, quantity, money);
                    return truncation(money);
                }).whenComplete((result, t) -> {
                    log.info("Completed: result: {}", t == null ? result :
                            (t.getClass().getName() + "." + t.getMessage()));
                }).get();
    }

    static double fallbackMoney() {
        return 1.0;
    }

    //这是一个consumer：只打印日志
    static void printAsLog(Object x) {
        log.info("{}", x);
    }

    //这是一个 Function，把入参 d 保留两位小数
    static double truncation(double d) {
        DecimalFormat format = new DecimalFormat("#.00");
        double rt = Double.parseDouble(format.format(d));
        log.info("{} truncation to {}", d, rt);
        return rt;
    }
}
